package com.song.boot.springstudy.io.nio.v3;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 主类，用于启动服务器并初始化线程组。
 */
public class SocketMultiplexingThreadsV2 {
    public static void main(String[] args) throws IOException {
        System.out.println(Thread.currentThread());

        // 创建Boss线程组，负责接收连接
        EventLoopGroup boss = new EventLoopGroup(1);
        // 创建Worker线程组，负责处理读写事件
        EventLoopGroup worker = new EventLoopGroup(3);
        ServerBootStrap b = new ServerBootStrap();
        // 绑定端口并启动服务器
        b.group(boss, worker).bind(9090);

        // 阻塞主线程，等待用户输入
        System.in.read();
    }
}

/**
 * 服务器启动引导类，负责绑定端口和启动ServerSocketChannel。
 */
class ServerBootStrap {
    private EventLoopGroup group;
    private EventLoopGroup chiledGroup;
    ServerAcceptr sAcceptr;

    /**
     * 设置Boss和Worker线程组。
     * @param boss Boss线程组，负责接收连接。
     * @param worker Worker线程组，负责处理读写事件。
     * @return 当前对象，支持链式调用。
     */
    public ServerBootStrap group(EventLoopGroup boss, EventLoopGroup worker) {
        group = boss;
        chiledGroup = worker;
        return this;
    }

    /**
     * 绑定端口并启动ServerSocketChannel。
     * @param port 监听的端口号。
     * @throws IOException 如果发生IO异常。
     */
    public void bind(int port) throws IOException {
        // 打开ServerSocketChannel并设置为非阻塞模式
        ServerSocketChannel server = ServerSocketChannel.open();
        server.configureBlocking(false);
        server.bind(new InetSocketAddress(port));
        sAcceptr = new ServerAcceptr(chiledGroup, server);

        // 选择一个EventLoop线程，将启动ServerSocketChannel的操作推送到该线程中执行
        EventLoop eventloop = group.chosser();
        eventloop.execute(new Runnable() {
            @Override
            public void run() {
                eventloop.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            // 将ServerSocketChannel注册到Selector上，监听ACCEPT事件
                            eventloop.name = Thread.currentThread() + eventloop.name;
                            System.out.println("bind...server...to " + eventloop.name);
                            server.register(eventloop.selector, SelectionKey.OP_ACCEPT, sAcceptr);
                        } catch (ClosedChannelException e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        });
    }
}

/**
 * 线程组类，管理一组EventLoop线程。
 */
class EventLoopGroup {
    AtomicInteger cid = new AtomicInteger(0);
    EventLoop[] childrens = null;

    /**
     * 构造函数，初始化线程组。
     * @param nThreads 线程组中线程的数量。
     */
    EventLoopGroup(int nThreads) {
        childrens = new EventLoop[nThreads];
        for (int i = 0; i < nThreads; i++) {
            childrens[i] = new EventLoop("T" + i);
        }
    }

    /**
     * 选择一个EventLoop线程。
     * @return 选择的EventLoop线程。
     */
    public EventLoop chosser() {
        return childrens[cid.getAndIncrement() % childrens.length];
    }
}

/**
 * 处理读取事件的接口。
 */
interface Handler {
    void doRead();
}

/**
 * 客户端读取器，负责处理客户端的读取事件。
 */
class ClientReader implements Handler {

    SocketChannel key;

    /**
     * 构造函数，初始化SocketChannel。
     * @param server SocketChannel对象。
     */
    ClientReader(SocketChannel server) {
        this.key = server;
    }

    /**
     * 读取客户端数据并回写。
     */
    @Override
    public void doRead() {
        ByteBuffer data = ByteBuffer.allocateDirect(4096);
        try {
            key.read(data);
            data.flip();
            byte[] dd = new byte[data.limit()];
            data.get(dd);
            System.out.println(new String(dd));
            data.clear();
            for (int i = 0; i < 10; i++) {
                data.put("a".getBytes());
                data.flip();
                key.write(data);
                data.clear();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

/**
 * 服务器接受器，负责处理连接接受事件。
 */
class ServerAcceptr implements Handler {
    ServerSocketChannel key;
    EventLoopGroup cGroup;

    /**
     * 构造函数，初始化ServerSocketChannel和Worker线程组。
     * @param cGroup Worker线程组。
     * @param server ServerSocketChannel对象。
     */
    ServerAcceptr(EventLoopGroup cGroup, ServerSocketChannel server) {
        this.key = server;
        this.cGroup = cGroup;
    }

    /**
     * 处理连接接受事件。
     */
    public void doRead() {
        try {
            // 选择一个Worker线程，将客户端连接注册到该线程的Selector上
            final EventLoop eventLoop = cGroup.chosser();
            final SocketChannel client = key.accept();
            client.configureBlocking(false);
            client.setOption(StandardSocketOptions.TCP_NODELAY, true);
            final ClientReader cHandler = new ClientReader(client);
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        eventLoop.name = Thread.currentThread() + eventLoop.name;
                        System.out.println("socket...send...to " + eventLoop.name);
                        client.register(eventLoop.selector, SelectionKey.OP_READ, cHandler);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

/**
 * EventLoop类，负责处理Selector上的事件。
 */
class EventLoop implements Executor {

    Selector selector;
    Thread thread = null;
    BlockingQueue events = new LinkedBlockingQueue();
    int NOT_STARTED = 1;
    int STARTED = 2;
    String name;

    /**
     * 构造函数，初始化Selector。
     * @param name 线程名称。
     */
    public EventLoop(String name) {
        try {
            this.name = name;
            selector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 主循环，处理Selector上的事件。
     * @throws InterruptedException 如果线程被中断。
     * @throws IOException 如果发生IO异常。
     */
    public void run() throws InterruptedException, IOException {
        for (; ; ) {
            // 调用Selector.select()方法，阻塞等待事件
            System.out.println("server已经开始：");
            int nums = selector.select();
            if (nums > 0) {
                // 获取已选择的SelectionKey集合
                Set<SelectionKey> keys = selector.selectedKeys();
                Iterator<SelectionKey> iter = keys.iterator();
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    iter.remove();
                    Handler handler = (Handler) key.attachment();
                    if (handler instanceof ServerAcceptr) {
                    } else if (handler instanceof ClientReader) {
                    }
                    handler.doRead();
                }
            }
            // 执行任务队列中的任务
            runrTask();
        }
    }

    /**
     * 执行任务。
     * @param task 要执行的任务。
     */
    @Override
    public void execute(Runnable task) {
        try {
            events.put(task);
            this.selector.wakeup();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (!inEventLoop()) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        thread = Thread.currentThread();
                        EventLoop.this.run();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }

    /**
     * 执行任务队列中的任务。
     * @throws InterruptedException 如果线程被中断。
     */
    public void runrTask() throws InterruptedException {
        for (int i = 0; i < 5; i++) {
            Runnable task = (Runnable) events.poll(10, TimeUnit.MILLISECONDS);
            if (task != null) {
                events.remove(task);
                task.run();
            }
        }
    }

    /**
     * 判断当前线程是否在EventLoop中运行。
     * @return 如果当前线程是EventLoop线程，则返回true；否则返回false。
     */
    private boolean inEventLoop() {
        return thread == Thread.currentThread();
    }
}